/* 
 * File:   parSolver.h
 * Author: tomas
 *
 * Created on May 12, 2009, 1:13 AM
 */

#ifndef _PARSOLVER_H
#define	_PARSOLVER_H

#include "ParStack.h"
#include "solution.h"

#include <cstdlib>
#include <ctime>
#include <set>

#include <boost/mpi.hpp>

namespace mpi = boost::mpi;

extern const bool DEBUG;

class ParSolver {
public:

    ParSolver(Instance& in, mpi::environment* _env, mpi::communicator* _world) :
	  instance(in), env(_env), world(_world), openNodes(&in), myRank(world->rank()) {
		
        // inicializace nah. cisel
        // k casu pridam rank, tim by se mohlo ovlivnit generovani nah. cisel,
        // aby kazdy proces mel trochu jinou inicializaci, pousti se ve stejny cas
        std::srand(time(NULL) + myRank*10);

        solutionLayout = StackRecord::boardLayout_t(instance.n() * instance.n());

        // na zacatku maji vsechny procesory bilou barvu
        processColor = WHITE;
        tokenColor = MSG_TOKEN_WHITE;

        requetAnotherWork = true;

        if(myRank==MASTER_RANK) {
            tokenPrisel = true; // zajisti ze master pole prniho peska
        } else {
            tokenPrisel = false;
        }


    }

    static const int MASTER_RANK = 0;

    static const unsigned CHECK_MSG_AMOUNT = 10;

    static const unsigned MSG_WORK_REQUEST = 1000;
    static const unsigned MSG_WORK_SENT = 1001;
    static const unsigned MSG_WORK_NOWORK = 1002; // nepouzivam, misto tohoto se posle prazdna cast zasobniku
    static const unsigned MSG_TOKEN = 1003;
    static const unsigned MSG_FINISH = 1004;

    static const unsigned MSG_TOKEN_WHITE = 10;
    static const unsigned MSG_TOKEN_BLACK = 11;

    enum ProcessColor {
        BLACK,
        WHITE
    };


    //////////////////////////////////////////////////////////////////////////

    virtual Solution solve() {
		mpi::timer myTimer;

        createSolLay();
		stateCounter = 0;
		requestCounter = 0;
		failedRequestCounter = 0;

        // pokud nejsi master, cekej az ti prijde prace
        if (myRank != MASTER_RANK) {

            // prijmi praci
			utils::log("ceka na praci (prvni rozdeleni)");

            // cekam az prijde nejaka zprava
            boost::optional<mpi::status> status =  world->probe(mpi::any_source, mpi::any_tag);

			utils::log("prisla message (prvni rozdeleni)");

            // prob-u muze byt za sebou vice... 
            processMessages();


        } else {
            // zacinam od nuly
            StackRecord::boardLayout_t start = createLayout(instance);
            openNodes.push(StackRecord(start)); //s prazdnou historii

            // priznak zpracovavani prvnih stavu pred prvnim rozeslanim prace
            processFirstDistribution = true;
        }

		utils::log("start vypoctu");

        citac = 0;
        exitProc = false;
        while (!exitProc)
		{
            if (!openNodes.empty())
			{
                processTopSolution();
                expandTop();
            }
			else
			{
                sendToken();
                requestWork();
            }
            //if ((citac % CHECK_MSG_AMOUNT) == 0  )
			if ((citac & 0x3ff) == 0  )
			{
                if (myRank == MASTER_RANK && processFirstDistribution)
				{
                    distributeBeginWork();
                }
				else
				{
                    int job = processMessages();
                    
                    if ( job == JOB_FINISH ) {

						utils::log("job finish");
                        // nevim jestli tu ma opravdu byt, zatim jen takovy pokus
                        // world->barrier();
                        //log("redukce");
						utils::log(utils::toString("redukce start ", myTimer.elapsed()));
						mpi::reduce(*world, tempBest, tempBest, mpi::minimum<Solution>(), MASTER_RANK );
						utils::log(utils::toString("redukce stop ", myTimer.elapsed()));
						world->barrier();
						utils::masterPrint(utils::toString(
							"Total time elapsed = ", myTimer.elapsed()
							));

						//posbiram citace prozkoumanych stavu
						std::vector<unsigned long long> stateCounts;
						mpi::gather<unsigned long long>(*world, stateCounter, stateCounts, MASTER_RANK);

						if (myRank == MASTER_RANK)
						{
							unsigned long long sum = 0;
							for (unsigned i = 0; i!=stateCounts.size(); ++i)
							{
								utils::print(utils::toString("States discovered at " , i, ": ", stateCounts[i]));
								sum += stateCounts[i];
							}
							utils::masterPrint(utils::toString("Total states count: ", sum));
						}
						utils::print(utils::toString(
							"Requests [total/failed]: ", requestCounter, "/", failedRequestCounter
							));
						world->barrier();
						return tempBest;
                    }
                }
            }
            citac++;
        }
    } //end of solve()

    //////////////////////////////////////////////////////////////////////////

    virtual ~ParSolver() {

    }


private:

    Instance instance;
    mpi::environment* env;
    mpi::communicator* world;

    ParStack openNodes;

    unsigned citac;
    StackRecord::boardLayout_t solutionLayout;
    Solution tempBest;
    unsigned long long stateCounter;
	unsigned long long requestCounter;
	unsigned long long failedRequestCounter;

    bool processFirstDistribution; // priznak zpracovavani prvnih stavu pred prvnim rozeslanim prace

    bool exitProc; // priznak ukonceni

    ProcessColor processColor;
    unsigned tokenColor;

    bool tokenPrisel;
    bool requetAnotherWork;

    //typedef std::map<int, int> workRequests_t;
	typedef std::set<int> workRequests_t;

    static const unsigned JOB_FINISH = 1;

	const int myRank;

    

    //////////////////////////////////////////////////////////////////////////

    unsigned processMessages() {

        workRequests_t workRequests;

        mpi::request r;
        WorkUnit work;

        int msg;
        int token;

        boost::optional<mpi::status> status;
        while (status = world->iprobe(mpi::any_source, mpi::any_tag)) {

            switch (status.get().tag()) {
                case MSG_WORK_REQUEST:

                    // klicem je adresa zdroje, hodnotou take
                    // pokud prijde vice zadosti od jednoho, bere se to porad jako jedna zadost
                    // odpovi se az pri zpracovani requestu, to se deje az po zpracovani vsech zprav, protoze dopredu bych nevedel na kolik zasobnik rozdelit
                    //workRequests[status.get().source()] = status.get().source();
					workRequests.insert(status->source());

                    r = world->irecv(status.get().source(), status.get().tag(), msg);
                    r.wait();
                    break;

                case MSG_WORK_SENT:
                    // prisel rozdeleny zasobnik
                    r = world->irecv(status.get().source(), status.get().tag(), work);
                    r.wait();

                    // pridam praci do zasobniku
                    openNodes.addWorkUnits(work);

                    utils::log(utils::toString(
						"prijata prace: size = ", work.u.size(), " od: ", status.get().source()
						));

                    requetAnotherWork = true;

                    break;

                case MSG_WORK_NOWORK:
					//TODO: asi nahradit za blokujici prijem.
                    r = world->irecv(status.get().source(), status.get().tag(), msg);
                    r.wait();
					failedRequestCounter++;
                    utils::log(utils::toString("zadna prace od: ", status.get().source()));

                    requetAnotherWork = true;

                    break;
                    
                case MSG_TOKEN:
                    r = world->irecv(status.get().source(), status.get().tag(), token);
                    r.wait();

                    // vsichni krome mastera musi cekat az prijde token, nez poslou svuj
                    tokenPrisel = true;

                    if(MSG_TOKEN_WHITE == token) {
						utils::log(utils::toString(
							"prijat token MSG_TOKEN_WHITE: od ", status.get().source()
							));
                    } else {
						utils::log(utils::toString(
							"prijat token MSG_TOKEN_BLACK: od ", status.get().source()
							));
                    }

                    utils::log(utils::toString("velikost zasobniku: ", openNodes.size()));

                    // PESEK
                    // Jestlize Pi obdrzi peska, pak ma-li Pi barvu B, nastavi barvu peska na B.
                    if (token != MSG_TOKEN_WHITE && token != MSG_TOKEN_BLACK) {
						utils::log("CHYBA - spatna hodnota tokenu");
                        throw ("spatna hodnotka tokenu!");
                    } else {
                        if (processColor == BLACK) {
                            tokenColor = MSG_TOKEN_BLACK;
							utils::log("token obarven na MSG_TOKEN_BLACK" );
                        } else {
                            tokenColor = MSG_TOKEN_WHITE;
                        }
                        //Pokud P1 obdrzi zpet peska barvy W, je mozne vypocet ukoncit. Pokud ma pesek barvu B, P1 zahaji nove kolo vyslanim barvy W procesoru P2.
                        if (myRank == MASTER_RANK) {
                            if(token == MSG_TOKEN_WHITE) {
                                // je mozne ukoncit vypocet
                                broadcast(MSG_FINISH);
                                return JOB_FINISH;
                            } else {
                                tokenColor = MSG_TOKEN_WHITE;
                            }
                        }
                    }
                    break;

                case MSG_FINISH:
                    //konec vypoctu - proces 0 pomoci tokenu zjistil, ze jiz nikdo nema praci
                    //a rozeslal zpravu ukoncujici vypocet
                    //mam-li reseni, odeslu procesu 0
                    //nasledne ukoncim spoji cinnost
                    //jestlize se meri cas, nezapomen zavolat koncovou barieru MPI_Barrier (MPI_COMM_WORLD)

                    return JOB_FINISH;

                    break;

                default:
					utils::log("CHYBA - neznamy typ zpravy");
                    throw ("neznamy typ zpravy");
                    break;
            }

        }

        distributeWork( workRequests );

        return 0;

    }


    //////////////////////////////////////////////////////////////////////////
    void broadcast(int msgType) {
		utils::log("broadcast ");
        for(int i=0; i<world->size(); i++) {
            if(i != myRank) {
                // sobe neposilam
                mpi::request r = world->isend(i, msgType, 0);
                r.wait();
            }
        }
    }


    //////////////////////////////////////////////////////////////////////////
    // distribuce dat mezi procesory, ktere chteji praci
    void distributeWork(const workRequests_t& workRequests) {

        if (!workRequests.empty())
		{
            // zpracuji pozadovanou praci
            // --- DISTRIBUCE DAT ---
            std::vector<WorkUnit> work = openNodes.divide(workRequests.size());
            utils::log(utils::toString("pocet requestu: ", workRequests.size()));

            int workIndex = 0;

			//for pres vsechny zadosti o praci
            for (workRequests_t::const_iterator it = workRequests.begin(); it != workRequests.end(); ++it)
			{
                //int dest = it->second;
				int dest = *it;

                if (workIndex < work.size()) {
                    // pokud mam jeste nejakou praci, poslu ji
                    mpi::request r = world->isend(dest, MSG_WORK_SENT, work.at(workIndex));
                    r.wait(); //pocka, nez skonci posilani dat, ale uz neblokuje do prijeti na cilove strane

					utils::log( utils::toString(
							"distribuce prace: pro: ",
							dest,
							" velikost ",
							work.at(workIndex).u.size()
							));

                    workIndex++;

                    // PESEK
                    // Jestlize procesor Pi posle praci procesoru Pj, kde i>j, pak Pi nastavi svou barvu na B.
                    if (myRank > dest) {
                        processColor = BLACK;
                    }

                } else {
                    // uz nemam co nabidnout, praci neposlu
                    mpi::request r = world->isend(dest, MSG_WORK_NOWORK, 0);
                    r.wait();
					utils::log(utils::toString("distribuce prace: pro:", dest, " nic"));
                }
            } // end of for
            utils::log("distribuce prace: konec");
            // ----------------------
        }

    }


    //////////////////////////////////////////////////////////////////////////
    void sendToken() {
        // token muzu poslat az kdyz obdrzim token od mastra, master posila jako 
        //if (tokenPrisel || myRank == MASTER_RANK) {
         processColor = WHITE;
        if (tokenPrisel) {
            // PESEK
            //Jakmile se prvni procesor P1 stane idle, nastavi svou barvu na W a posle peska barvy W procesoru P2.
            //Jakmile se Pi stane idle, posle peska po kruznici procesoru Pi+1 a nastavi svoji barvu na W.
            int dest = (myRank + 1) % world->size();
            mpi::request r = world->isend(dest, MSG_TOKEN, tokenColor);
            r.wait();

            tokenPrisel = false;

            if(tokenColor == MSG_TOKEN_WHITE) {
                utils::log(utils::toString(
					"odeslan token MSG_TOKEN_WHITE pro:", dest
					));
            } else {
                utils::log(utils::toString(
					"odeslan token MSG_TOKEN_BLACK pro:", dest
					));
            }
        }
    }

    //////////////////////////////////////////////////////////////////////////
    // deli praci po pocatecnim napocitani stavu
    
    void distributeBeginWork() {
		utils::log("distribute begin work");
        
        workRequests_t workRequests;

        // nasimuluju si requesty pro cekajici procesy
        for(int i=1; i<world->size(); i++) {
            //workRequests[i] = i;
			workRequests.insert(i);
        }

        // distribuji praci
        distributeWork(workRequests);

        // prvni prace byla rozeslana
        processFirstDistribution = false;

    }

    //////////////////////////////////////////////////////////////////////////

    void requestWork() {

        //        //nahodne cislo v boost
        //        boost::mt19937 rng; // typ pseudo generatoru
        //        boost::uniform_int<> sestka(1, 6) // distribuce hodnot 1 az 6
        //
        //        boost::variate_generator<boost::mt19937&, boost::uniform_int<> >kostka(rng, sestka); // spojeni generatoru a distribuce => kostka6
        //        int x = kostka(); // jeden hod kostkou

        // nebo cist z /dev/random

        if(requetAnotherWork) {
            requetAnotherWork = false;

            // vybrani nahodneho darce (ne sebe)
            int dest = rand() % world->size();
            while(dest == myRank) {
                dest = rand() % world->size();
            }

            utils::log(utils::toString("zadost o praci pro:", dest));
            // zadost o praci
            mpi::request r = world->isend(dest, MSG_WORK_REQUEST, 0);
            r.wait();
			requestCounter++;
        }
	} //end of requestWork()

    //////////////////////////////////////////////////////////////////////////
    void createSolLay() {
        size_t cnt = instance.k();
        size_t y = 0, x = 0; //zacnu nahore vlevo
        size_t zarazka = instance.n();
        while (cnt > 0) {
            size_t index = utils::coordsToIndex(x, y, instance.n());

            //solutionLayout[index]=1;
            solutionLayout.setAt(index, true);
            x++;
            if (x >= zarazka) {
                x = 0; //zpet na zacatek radku
                y++; //po radcich dolu
                zarazka--;
            }
            cnt--;
        }
    }

    //////////////////////////////////////////////////////////////////////////
    void processTopSolution() {
        if (isTopSolution()) { //na vrcholu zasobniku je reseni!
            StackRecord top = openNodes.top();
            openNodes.pop(); //z reseni uz nebude lepsi.. leda bychom meli zaporne penalizace (coz nemame)
			//lepsi je reseni s mensi penalizaci
            if (tempBest.penaltySum > top.totalPenalty) {
                //nasli jsme lepsi reseni!
				utils::log(utils::toString("Velikost zasobniku: ", openNodes.size() ));
                utils::log(utils::toString("Pocet prohledanych stavu: ", this->stateCounter));
                utils::log(utils::toString("SOLUTION:\n", top));
                tempBest = Solution(top.movesHistory, top.totalPenalty);
            }
        }
        //pokud neni reseni, nic se nestane.
    }

    //////////////////////////////////////////////////////////////////////////
	// Test jestli je na vrcholu zasobniku reseni
    bool isTopSolution() {
        return openNodes.top().boardLayout == solutionLayout;
    }

    //////////////////////////////////////////////////////////////////////////
	// Metoda vyjme vrsek zasobniku a vlozi na nej jeho nasledniky
	// TODO: provest orezani apriori, tedy pred vlozenim stavu na zasobnik misto
	// pri expanzi vrcholu.
    void expandTop() {
        if (!openNodes.empty()) {
            StackRecord actual = openNodes.top();
            //std::cerr << actual << std::endl;
            openNodes.pop();
            this->stateCounter++;
            //max hloubka:
            if (actual.depth == instance.q()) {
                return;
            }
            //if (actual.totalPenalty >= tempBest.penaltySum) { //z tohoto uzlu uz nebude lepsi reseni nez jaky jsem uz nasel. oriznout..
            //    return;
            //}
            utils::coords_t * origCp = 0;
            if (!actual.movesHistory.empty()) {
                origCp = &(actual.movesHistory.back().origCoords);
            }
            for (size_t i = 0; i < actual.boardLayout.size(); ++i) {
                if (!actual.boardLayout[i]) {
                    continue;
                }
                for (size_t j = 1; j <= 8; ++j) {
					//genetate all possible moves
                    Move m(utils::indexToCoords(i, instance.n()), (Direction) j);
					//exclude direct move back to where we came from.
                    if (origCp && m.targetCoords == *origCp) {
                        continue;
                    }
                    StackRecord::boardLayout_t newL = move(actual.boardLayout, m);
                    if (newL == actual.boardLayout) { //Pokud je tah neplatny, vratilo mi to nezmenenou konfiguraci
                        continue;
                    }
                    size_t penalty = instance.getPenalty(m.targetCoords);
					if (actual.totalPenalty+penalty >= tempBest.penaltySum)
					{ //orezani podle penalizace, pred ulozenim na zasobnik
						stateCounter++; //i tyto stavy bych mel pocitat, ptz.jsem u nich stravil dost vypoctu
						continue;
					}
                    openNodes.push(StackRecord(newL, actual, m, penalty));
                }
            }
        }
    } //end of expandTop()

    //returns the same layout, if the move would be invalid.
    StackRecord::boardLayout_t move(const StackRecord::boardLayout_t &_inLay, const Move& _m) {
        utils::coords_t new_coor = _m.targetCoords;
        size_t new_index;
        if (new_coor.first >= instance.n() || //kontrola souradnic
                new_coor.second >= instance.n() ||
                _inLay[(new_index = utils::coordsToIndex(new_coor, instance.n()))] //test cilove pozice, jestli je volna
                ) {
            return _inLay; //v pripade ze tento tah neni platny, vratim puvodni rozlozeni
        }
        StackRecord::boardLayout_t new_lay = _inLay;
        new_lay.setAt(utils::coordsToIndex(_m.origCoords, instance.n()), false);
        new_lay.setAt(new_index, true);

        return new_lay;
    }

    //Metoda pro vytvoreni bitmapy sachovnice pro dane zadani.
    static StackRecord::boardLayout_t createLayout(Instance &_inst) {
        StackRecord::boardLayout_t lay(_inst.n() * _inst.n()); //velikost n*n !!
        for (size_t i = 0; i < _inst.getHorses().size(); ++i) {
            size_t x = _inst.getHorses()[i].x;
            size_t y = _inst.getHorses()[i].y;
            size_t index = utils::coordsToIndex(x, y, _inst.n());
            lay.setAt(index, true);
        }
        return lay;
    }

};



#endif	/* _PARSOLVER_H */

